package com.qingyun.pdfservice.utils;


import com.qingyun.pdfservice.service.IPdfService;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.qingyun.common.util.RedisConstants.*;



/**
 * PDF专用 Redis Stream 消费者 —— 多线程版
 */
@Component
public class StreamMessageConsumer {

    private final RedisTemplate<String, Object> redisTemplate;
    private final IPdfService pdfService;

    //单线程池
    // private final ExecutorService executorService = Executors.newSingleThreadExecutor();

    // 多线程池
    private static final int CONSUMER_THREADS = Runtime.getRuntime().availableProcessors() << 1;
    private final ExecutorService consumerPool = Executors.newFixedThreadPool(CONSUMER_THREADS);

    public StreamMessageConsumer(RedisTemplate<String, Object> redisTemplate, IPdfService pdfService) {
        this.redisTemplate = redisTemplate;
        this.pdfService = pdfService;
        initConsumerGroup();
        startConsuming();
    }

    private void initConsumerGroup() {
        try {
            redisTemplate.opsForStream().createGroup(
                    PDF_TASK_QUEUE,
                    ReadOffset.latest(),
                    PDF_TASK_GROUP
            );
        } catch (Exception ignore) {}
    }

    /*
    public void startConsuming() {
        executorService.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read(
                            Consumer.from(PDF_TASK_GROUP, PDF_TASK_CONSUMER),
                            StreamReadOptions.empty().block(Duration.ofSeconds(2)).count(10),
                            StreamOffset.create(PDF_TASK_QUEUE, ReadOffset.lastConsumed())
                    );
                    if (records == null || records.isEmpty()) continue;

                    for (MapRecord<String, Object, Object> r : records) {
                        try {
                            Map<Object, Object> map = r.getValue();
                            Object fid = map.get(FIELD_FEEDBACK_ID);
                            if (fid != null) {
                                Long feedbackId = Long.valueOf(String.valueOf(fid));
                                pdfService.generateAndSaveFeedbackPdf(feedbackId);
                            }
                            redisTemplate.opsForStream().acknowledge(PDF_TASK_GROUP, r);
                        } catch (Exception ex) {
                            // 不ack，进入pending
                        }
                    }
                } catch (Exception e) {
                    try { Thread.sleep(500L); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); }
                }
            }
        });
    }*/


    //多线程消费逻辑：主线程拉消息，工作线程并行处理
    public void startConsuming() {
        //只用 1 条后台线程去不断拉取，防止连接竞争
        Executors.newSingleThreadExecutor().submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read(
                            Consumer.from(PDF_TASK_GROUP, PDF_TASK_CONSUMER),
                            StreamReadOptions.empty().block(Duration.ofSeconds(2)).count(10),
                            StreamOffset.create(PDF_TASK_QUEUE, ReadOffset.lastConsumed())
                    );
                    if (records == null || records.isEmpty()) continue;

                    /* 一批消息分发给多条线程 */
                    CountDownLatch latch = new CountDownLatch(records.size());
                    for (MapRecord<String, Object, Object> r : records) {
                        consumerPool.execute(() -> {
                            try {
                                Map<Object, Object> map = r.getValue();
                                Object fid = map.get(FIELD_FEEDBACK_ID);
                                if (fid != null) {
                                    Long feedbackId = Long.valueOf(String.valueOf(fid));
                                    pdfService.generateAndSaveFeedbackPdf(feedbackId);
                                }
                                /* 业务成功后再 ack */
                                redisTemplate.opsForStream().acknowledge(PDF_TASK_GROUP, r);
                            } catch (Exception ex) {
                                // 不 ack，消息进入 pending-list，后续可重试
                            } finally {
                                latch.countDown();
                            }
                        });
                    }
                    /* 主线程可继续去拉，不等待这批做完 */
                } catch (Exception e) {
                    try { Thread.sleep(500L); } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        });
    }
}